package com.study.flink.java.day01_wc;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * 流式处理单词计数word count例子
 * @author linys
 */
public class StreamWordCount {
    public static void main(String[] args) throws Exception {
        // 获取上下文环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.source，从外部socket读取流数据
        DataStream<String> lines = env.socketTextStream("node02", 8888);

        // 2.transformation
        SingleOutputStreamOperator<String> word = lines.flatMap(
                (String line, Collector<String> out) -> Arrays.stream(line.split(","))
                        .forEach(out::collect)).returns(Types.STRING);
        SingleOutputStreamOperator<String> wordFilter = word.filter(
                s -> !"".equals(s)).returns(Types.STRING);
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = wordFilter.map(
                w -> Tuple2.of(w, 1)).returns(Types.TUPLE(Types.STRING, Types.INT));

        // 分组
        KeyedStream<Tuple2<String, Integer>, Tuple> grouped = wordAndOne.keyBy(0);
        // 聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = grouped.sum(1);

        // 3.sink，打印
        summed.print();
        // 执行
        env.execute("StreamWordCount");
    }
}
